package cn.chenchendemospringboot.stock_demo.schedule;

import cn.chenchendemospringboot.stock_demo.entity.KafkaEventProcess;
import cn.chenchendemospringboot.stock_demo.mapper.KafkaEventProcessMapper;
import cn.chenchendemospringboot.stock_demo.utils.KafkaUtils;
import com.google.common.util.concurrent.ThreadFactoryBuilder;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import javax.annotation.PostConstruct;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ThreadFactory;


/**
 * @author Dean
 * 项目启动  接收事件  入库
 */
@Service
public class EventProcessService {

    @Autowired
    KafkaEventProcessMapper kafkaEventProcessMapper;

    @PostConstruct
    public void init() {
        ThreadFactory threadFactory = new ThreadFactoryBuilder()
                .setNameFormat("MqMessageConsumeThread-%d")
                .setDaemon(true)
                .build();
        ExecutorService executorService = Executors.newSingleThreadExecutor(threadFactory);
        executorService.execute(new MqMessageConsumeThread());
    }

    private class MqMessageConsumeThread implements Runnable {
        @Override
        public void run() {
            KafkaUtils.consume(consumerRecord -> {
                KafkaEventProcess eventProcess = new KafkaEventProcess();
                eventProcess.setPayload(consumerRecord.value());
                eventProcess.setEventType("EVENTPUBLISH_CREATE");
                eventProcess.setStatus("NEW");
                kafkaEventProcessMapper.insert(eventProcess);
                System.out.println("保存保存");
            });
        }
    }
}

